package com.pass.three;

import java.io.IOException;
import java.util.HashSet;
import java.util.Set;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import com.data.ConnectedEdge;
import com.data.Node;
import com.join.SpatialJoin;
import com.uf.UnionFind;
import com.utils.CommonUtils;

/**
 * Pass Three Reducer gets Nodes in the group, does spatial join, then does
 * union find with all edges. Write <GroupID, edge edge edge..> to file. Done.
 */
public class PassThreeReducer extends
		Reducer<IntWritable, Text, IntWritable, Text> {
	@Override
	protected void reduce(IntWritable key, Iterable<Text> values,
			Context context) throws IOException, InterruptedException {
		
		Set<ConnectedEdge> edges = new HashSet<ConnectedEdge>();
		for (Text value : values) {
			edges.addAll(CommonUtils.getEdgesFromString(value.toString()));
		}
		Set<ConnectedEdge> nodeEdges = new HashSet<ConnectedEdge>();
		Set<Node> nodes = new HashSet<Node>();
		for (ConnectedEdge edge : edges) {
			if (edge.isNode()) {
				nodeEdges.add(edge);
				nodes.add(edge.getNode1());
			}
		}
		edges.removeAll(nodeEdges);

		// spatial join on nodes --> edges
		if (nodes.size() > 0) {
			Set<ConnectedEdge> edgesBeforeUF = SpatialJoin.spatialJoin(nodes);
			edges.addAll(edgesBeforeUF);
		}

		// run union find on all edges
		UnionFind uf = new UnionFind(edges);
		Set<ConnectedEdge> ufEdges = uf.getUFEdges();

		for (ConnectedEdge edge : ufEdges) {
			context.write(new IntWritable(0), new Text(edge.toString()));
		}
	}
}
